package main

import (
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

//计数器接口
type Counter interface {
	Increment()
	AddValue(int64)
	GetValue() int64
}

//QPS 计数器
type QPSCounter struct {
	count int64
}

func (c *QPSCounter) Increment() {
	atomic.AddInt64(&c.count, 1)
}
func (c *QPSCounter) AddValue(v int64) {
	atomic.AddInt64(&c.count, int64(v))
}
func (c *QPSCounter) GetValue() int64 {
	return atomic.LoadInt64(&c.count)
}
func NewQPSCount() Counter {
	return &QPSCounter{}
}

//时间生成器
type Timer struct {
}

func (t *Timer) getCurrentTimeInMillis() int64 {
	return time.Now().UnixNano() / 1e6
}

//桶
type Bucket struct {
	windowStart int64 //桶开始时间， 桶允许的时间长度为 windowStart 至 windowStart +  bucketSizeInMillseconds
	counter     Counter
}

//滑动窗口结构体
type RollingNumber struct {
	time                    Timer     //毫秒时间生成器
	timeInMilliseconds      int64     //时间窗口大小(毫秒)
	numberOfBuckets         int64     //桶的个数
	bucketSizeInMillseconds int64     //每个桶的时间长度(毫秒)
	buckets                 []*Bucket //
	front                   int64     //前指针
	rear                    int64     // 尾指针
	mux                     sync.RWMutex
	counter                 func() Counter //计数器生成函数
}

//生成桶
func (r *RollingNumber) newBucket(currTime int64, counter Counter) *Bucket {
	return &Bucket{windowStart: currTime, counter: counter}
}

//重置
func (r *RollingNumber) reset() {
	r.buckets = make([]*Bucket, r.numberOfBuckets)
	r.front = 0
	r.rear = 0
}

//获取最后一个桶
func (r *RollingNumber) peekLast() *Bucket {
	return r.buckets[r.rear%r.numberOfBuckets]
}

func (r *RollingNumber) addLast(bucket *Bucket) {

	r.clearBuckets(bucket.windowStart + r.bucketSizeInMillseconds)
	if r.buckets[r.rear%r.numberOfBuckets] != nil {
		r.rear++
	}
	if r.buckets[r.rear%r.numberOfBuckets] != nil {
		fmt.Println("old bucket----------", r.buckets[r.rear%r.numberOfBuckets], r.buckets[r.rear%r.numberOfBuckets].counter.GetValue())
	}

	r.buckets[r.rear%r.numberOfBuckets] = bucket

}

//清空滑动窗口的桶数据
func (r *RollingNumber) clearBuckets(currentTime int64) {
	if currentTime <= 0 {
		currentTime = r.time.getCurrentTimeInMillis()
	}
	println("clear time:", currentTime)
	for {
		if r.buckets[r.front%r.numberOfBuckets] == nil {
			println("clear nil bucket:", r.front%r.numberOfBuckets)
			atomic.AddInt64(&r.front, 1)
		} else if currentTime-(r.buckets[r.front%r.numberOfBuckets].windowStart+r.bucketSizeInMillseconds) >= r.timeInMilliseconds {
			println("clear overtime bucket:", r.front%r.numberOfBuckets, r.buckets[r.front%r.numberOfBuckets].windowStart)
			atomic.AddInt64(&r.front, 1)
		} else {
			break
		}
		if atomic.LoadInt64(&r.front) > atomic.LoadInt64(&r.rear) {
			println("clear all")
			r.reset()
			break
		}
	}
	println("clear time:", currentTime, "end")
}

//获取窗口计数
func (r *RollingNumber) getWindowCount() int64 {
	var ct int64
	r.mux.RLock()
	defer r.mux.RUnlock()
	r.clearBuckets(0)
	index := r.front
	fmt.Println(r.front, r.rear)
	for {
		if index > r.rear {
			break
		}
		if r.buckets[index%r.numberOfBuckets] != nil {
			ct += r.buckets[index%r.numberOfBuckets].counter.GetValue()
			fmt.Println("win count:", index, r.buckets[index%r.numberOfBuckets].counter.GetValue(), ct)
		}
		index++
	}
	return ct
}

//自增 1
func (r *RollingNumber) Increment() {
	r.mux.Lock()
	defer r.mux.Unlock()
	bucket := r.getCurrentBucket()
	bucket.counter.Increment()
	fmt.Println("increment -------------", bucket, bucket.counter.GetValue())
}

//增加 Value
func (r *RollingNumber) AddValue(value int64) {
	r.mux.Lock()
	defer r.mux.Unlock()
	bucket := r.getCurrentBucket()
	bucket.counter.AddValue(value)
}

//获取当前Bucket
func (r *RollingNumber) getCurrentBucket() *Bucket {
	currentTime := r.time.getCurrentTimeInMillis()
	lastBucket := r.peekLast()
	//当前时间还在最后一个桶的时间的时间范围内
	if lastBucket != nil && currentTime < lastBucket.windowStart+r.bucketSizeInMillseconds {
		return lastBucket
	}
	//如果最后一个桶是空，生成一个
	if lastBucket == nil {
		newBucket := r.newBucket(currentTime, r.counter())
		r.buckets[r.rear] = newBucket
		return newBucket
	}
	//如果当前时间 - 最后一个桶时间 大于 窗口时间，需要清空
	if currentTime-(lastBucket.windowStart+r.bucketSizeInMillseconds) > r.timeInMilliseconds {
		r.reset()
		return r.getCurrentBucket()
	}
	//当前时间大于最后一个桶的时间，要创建一个新的桶
	r.addLast(r.newBucket(lastBucket.windowStart+r.bucketSizeInMillseconds, r.counter()))

	return r.peekLast()

}
func NewRollingNumber(time Timer, timeInMilliseconds int64, numberOfBuckets int64) (*RollingNumber, error) {
	RM := &RollingNumber{
		time:               time,
		timeInMilliseconds: timeInMilliseconds,
		numberOfBuckets:    numberOfBuckets,
	}
	if timeInMilliseconds%numberOfBuckets != 0 {
		return nil, errors.New("The timeInMilliseconds must divide equally into numberOfBuckets. For example 1000/10 is ok, 1000/11 is not")
	}
	RM.bucketSizeInMillseconds = timeInMilliseconds / numberOfBuckets
	RM.buckets = make([]*Bucket, numberOfBuckets)
	fmt.Println(timeInMilliseconds, numberOfBuckets, RM.bucketSizeInMillseconds, RM.buckets)
	RM.counter = func() Counter {
		return NewQPSCount()
	}
	return RM, nil
}

func main() {
	r, err := NewRollingNumber(Timer{}, 100, 5)
	if err != nil {
		fmt.Println("NewRollingNumber error:", err)
	}
	wg := sync.WaitGroup{}
	for i := 0; i < 3; i++ {
		wg.Add(1)
		go func() {
			for j := 0; j < 10; j++ {
				r.Increment()
				time.Sleep(time.Millisecond * 10)
			}
			wg.Done()
		}()
	}
	wg.Wait()
	fmt.Println(r.buckets, r.getWindowCount())

}
